package com.sf.sgs.mark.exception.common.utils;

import com.sf.kafka.api.produce.IKafkaProducer;
import com.sf.kafka.api.produce.ProduceConfig;
import com.sf.kafka.api.produce.ProducerPool;

/**
 * Created by 01204808 on 2017/7/14.
 */
public class KafkaUtils {

    public static void sendMessage(String url, String topic, String token, String clusterName, String message) {
        final int poolSize = 5;
        // KAFKA连接地址
        // 主题所在的集群名称
        // 主题名称＋分隔符 （固定不变）＋主题的校验码
        String topicTokens = topic + ":" + token;
        // 简易模式
        ProduceConfig produceConfig = new ProduceConfig(poolSize, url, clusterName, topicTokens);
        IKafkaProducer kafkaProducer = new ProducerPool(produceConfig);
        kafkaProducer.sendString(topic, message);
    }
}
